#include "storage/buffer.h"
#include "common/hashfn.h"
KPHCPciPageBuffCtx *gPciBufCtx = NULL;

static inline uint32 KPHCPciHashCode(KPHCCfsBufferKey *key)
{
    return gPciBufCtx->hashtbl.hash((void *)key, sizeof(KPHCCfsBufferKey));
}

static inline uint32 KPHCPciKeyCmp(KPHCCfsBufferKey *keya, KPHCCfsBufferKey *keyb)
{
    return (uint32)gPciBufCtx->hashtbl.match((void *)keya, (void *)keyb, sizeof(KPHCCfsBufferKey));
}

KPHCPciPageCtrl *KPHCPciBufFindFromBucket(KPHCPciHashBucket *bucket, KPHCCfsBufferKey *key)
{
    KPHCPciPageCtrl *ctrl = NULL;
    uint32 ctrlId = bucket->firstCtrlId;
    while (ctrlId != PCI_INVALID_ID) {
        ctrl = PCI_GET_CTRL_BY_ID(gPciBufCtx, ctrlId);
        if (KPHCPciKeyCmp(key, &ctrl->pciKey) == 0) {
            return ctrl;
        }
        ctrlId = ctrl->bckNext;
    }
    return NULL;
}

void KPHCPciBufRemoveFromBucket(KPHCPciHashBucket *bucket, KPHCPciPageCtrl *item)
{
    KPHCPciPageCtrl *prev = NULL;
    KPHCPciPageCtrl *next = NULL;
    if (item->bckPrev != PCI_INVALID_ID) {
        prev = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->bckPrev);
        prev->bckNext = item->bckNext;
    } else {
        bucket->firstCtrlId = item->bckNext;
    }
    if (item->bckNext != PCI_INVALID_ID) {
        next = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->bckNext);
        next->bckPrev = item->bckPrev;
    }
    item->bckPrev = PCI_INVALID_ID;
    item->bckNext = PCI_INVALID_ID;
    item->bckId = PCI_INVALID_ID;
    memset(&(item->pciKey), 0, sizeof(item->pciKey));
    bucket->ctrlCount--;
}

void KPHCPciBufAddIntoBucket(KPHCPciHashBucket *bucket, KPHCPciPageCtrl *item)
{
    if (bucket->ctrlCount == 0) {
        item->bckId = bucket->bckId;
        item->bckPrev = PCI_INVALID_ID;
        item->bckNext = PCI_INVALID_ID;
        bucket->firstCtrlId = item->ctrlId;
        bucket->ctrlCount++;
        return;
    }

    KPHCPciPageCtrl *head = PCI_GET_CTRL_BY_ID(gPciBufCtx, bucket->firstCtrlId);
    item->bckId = bucket->bckId;
    item->bckPrev = PCI_INVALID_ID;
    item->bckNext = head->ctrlId;

    head->bckPrev = item->ctrlId;

    bucket->firstCtrlId = item->ctrlId;
    bucket->ctrlCount++;
}

void KPHCPciLruPushNoLock(KPHCPciLruList *lru, KPHCPciPageCtrl *item, KPHCCtrlState state)
{
    KPHCPciPageCtrl *temp = NULL;
    if (lru->count == 0) {
        item->lruNext = PCI_INVALID_ID;
        item->lruPrev = PCI_INVALID_ID;

        lru->first = item->ctrlId;
        lru->last = item->ctrlId;
        lru->count++;
        item->state = state;
        return ;
    }

    item->lruNext = lru->first;
    item->lruPrev = PCI_INVALID_ID;

    temp = PCI_GET_CTRL_BY_ID(gPciBufCtx, lru->first);
    temp->lruPrev = item->ctrlId;

    lru->first = item->ctrlId;
    lru->count++;
    item->state = state;
}

void KPHCPciLruPush(KPHCPciLruList *lru, KPHCPciPageCtrl *item, KPHCCtrlState state)
{
    (void)LWLockAcquire(lru->lock, LW_EXCLUSIVE);
    KPHCPciLruPushNoLock(lru, item, state);
    LWLockRelease(lru->lock);
}

void KPHCPciLruRemove(KPHCPciLruList *lru, KPHCPciPageCtrl *item)
{
    KPHCPciPageCtrl *prev = NULL;
    KPHCPciPageCtrl *next = NULL;
    if (item->lruPrev != PCI_INVALID_ID) {
        prev = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->lruPrev);
        prev->lruNext = item->lruNext;
    } else {
        lru->first = item->lruNext;
    }

    if (item->lruNext != PCI_INVALID_ID) {
        next = PCI_GET_CTRL_BY_ID(gPciBufCtx, item->lruNext);
        next->lruPrev = item->lruPrev;
    } else {
        lru->last = item->lruPrev;
    }

    item->lruPrev = PCI_INVALID_ID;
    item->lruNext = PCI_INVALID_ID;
    item->state = CTRL_STATE_INIT;
    lru->count--;

    return ;
}

KPHCPciPageCtrl *KPHCPciLruPop(KPHCPciLruList *lru)
{
    KPHCPciPageCtrl *item = NULL;
    if (lru->count == 0) {
        return NULL;
    }
    (void)LWLockAcquire(lru->lock, LW_EXCLUSIVE);
    if (lru->count == 0) {
        LWLockRelease(lru->lock);
        return NULL;
    }
    item = PCI_GET_CTRL_BY_ID(gPciBufCtx, lru->last);
    KPHCPciLruRemove(lru, item);
    LWLockRelease(lru->lock);
    return item;
}

bool KPHCPciBufRecycleCoreTry(KPHCPciPageCtrl *item)
{
    if ((pg_atomic_read_u32(&item->refNum) > 0) || (pg_atomic_read_u32(&item->touchNum) >= 3)) {
        return false;
    }

    KPHCPciHashBucket *bucket = PCI_GET_BUCKET_BY_ID(gPciBufCtx, item->bckId);
    if (!LWLockConditionalAcquire(bucket->lock, LW_EXCLUSIVE)) {
        return false;
    }
    if ((pg_atomic_read_u32(&item->refNum) > 0) || (pg_atomic_read_u32(&item->touchNum) >= 3)) {
        LWLockRelease(bucket->lock);
        return false;
    }
    KPHCPciBufRemoveFromBucket(bucket, item);
    pg_atomic_write_u32(&item->touchNum, 0);
    LWLockRelease(bucket->lock);
    return true;
}

KPHCPciPageCtrl *KPHCPciBufRecycleCore(KPHCPciLruList *mainLru)
{
    uint32 step = 0;
    KPHCPciPageCtrl *item = NULL;
    KPHCPciPageCtrl *res = NULL;
    if (!LWLockConditionalAcquire(mainLru->lock, LW_EXCLUSIVE)) {
        return res;
    }
    uint32 currCtrlId = mainLru->last;
    while (currCtrlId != PCI_INVALID_ID) {
        if (step >= 1024) {
            break;
        }
        item = PCI_GET_CTRL_BY_ID(gPciBufCtx, currCtrlId);
        if (KPHCPciBufRecycleCoreTry(item)) {
            KPHCPciLruRemove(mainLru, item);
            res = item;
            break;
        }
        //item->touchNum /= 2;
        int a = pg_atomic_read_u32(&item->touchNum);
        pg_atomic_sub_fetch_u32(&item->touchNum, 0.5 * a);
        step++;
        currCtrlId = item->lruPrev;
    }
    LWLockRelease(mainLru->lock);
    return res;
}

KPHCPciPageCtrl *KPHCPciBufRecycle(uint32 randVal)
{
    KPHCPciPageCtrl *item = NULL;
    uint8 randStart = randVal % PCI_PART_LIST_NUM;
    while (1) {
        for (uint8 i = 0; i < PCI_PART_LIST_NUM; i++) {
            item = KPHCPciLruPop(&gPciBufCtx->freeLru[(randStart + i) % PCI_PART_LIST_NUM]);
            if (item != NULL) {
                return item;
            }
        }
        for (uint8 i = 0; i < PCI_PART_LIST_NUM; i++) {
            item = KPHCPciBufRecycleCore(&gPciBufCtx->mainLru[(randStart + i) % PCI_PART_LIST_NUM]);
            if (item != NULL) {
                return item;
            }
            if (gPciBufCtx->freeLru[i].count != 0) {
                break;
            }
        }
    }
}


void KPHCPciBufLoadPage(KPHCPciPageCtrl *item, const KPHCExtentLocation location, KPHCCfsBufferKey *key)
{
    memcpy(&item->pciKey, key, sizeof(KPHCCfsBufferKey));
    int nbytes = FileRead(location.fd, (char *)item->pciPage, BLCKSZ,
                           location.headerNum * BLCKSZ, (uint32)WAIT_EVENT_DATA_FILE_READ);
    if (nbytes != BLCKSZ) {
        item->loadStatus = CTRL_PAGE_LOADED_ERROR;
        ereport(DEBUG5, (errcode(ERRCODE_DATA_CORRUPTED),
            errmsg("Failed to pci_buf_load_page %s, headerNum: %u.", FilePathName(location.fd), location.headerNum)));
        return;
    }
    item->loadStatus = CTRL_PAGE_IS_LOADED;
    return;
}

KPHCPciPageCtrl *KPHCPciBufReadPage(const KPHCExtentLocation location, LWLockMode lockMode, KPHCPciBufferReadMode readMode)
{
    KPHCCfsBufferKey key = {{location.relFileNode.spcNode, location.relFileNode.dbNode, location.relFileNode.relNode,
                         location.relFileNode.bucketNode},
        location.extentNumber};
    uint32 hashcode = KPHCPciHashCode(&key);
    KPHCPciHashBucket *bucket = PCI_GET_BUCKET_BY_HASH(gPciBufCtx, hashcode);
    (void)LWLockAcquire(bucket->lock, LW_EXCLUSIVE);
    KPHCPciPageCtrl *ctrl = KPHCPciBufFindFromBucket(bucket, &key);
    if (ctrl != NULL) {
        (void)LWLockAcquire(ctrl->contentLock, lockMode);
        if (ctrl->loadStatus != CTRL_PAGE_IS_LOADED && readMode == PCI_BUF_NORMAL_READ) {
            KPHCPciBufLoadPage(ctrl, location, &key);
        }
        (void)pg_atomic_fetch_add_u32(&ctrl->refNum, 1);
        (void)pg_atomic_fetch_add_u32(&ctrl->touchNum, 3);
        LWLockRelease(bucket->lock);
        return ctrl;
    }
    LWLockRelease(bucket->lock);
    KPHCPciPageCtrl *item = KPHCPciBufRecycle(location.extentNumber);
    (void)LWLockAcquire(bucket->lock, LW_EXCLUSIVE);
    ctrl = KPHCPciBufFindFromBucket(bucket, &key);
    if (ctrl != NULL) {
        (void)LWLockAcquire(ctrl->contentLock, lockMode);
        if (ctrl->loadStatus != CTRL_PAGE_IS_LOADED && readMode == PCI_BUF_NORMAL_READ) {
            KPHCPciBufLoadPage(ctrl, location, &key);
        }
        (void)pg_atomic_fetch_add_u32(&ctrl->refNum, 1);
        (void)pg_atomic_fetch_add_u32(&ctrl->touchNum, 1);
        LWLockRelease(bucket->lock);
        KPHCPciLruPush(&gPciBufCtx->freeLru[item->ctrlId % PCI_PART_LIST_NUM], item, CTRL_STATE_FREE);
        return ctrl;
    }
    pg_atomic_write_u32(&item->refNum, 1);
    pg_atomic_write_u32(&item->touchNum, 3);
    memset((void*)item->pciPage, 0, BLCKSZ);
    item->loadStatus = CTRL_PAGE_NO_LOAD;
    if (readMode == PCI_BUF_NORMAL_READ) {
        KPHCPciBufLoadPage(item, location, &key);
    }
    KPHCPciBufAddIntoBucket(bucket, item);
    (void)LWLockAcquire(item->contentLock, lockMode);
    LWLockRelease(bucket->lock);
    KPHCPciLruPush(&gPciBufCtx->mainLru[item->ctrlId % PCI_PART_LIST_NUM], item, CTRL_STATE_USED);

    return item;
}

void KPHCPciBufFreePage(KPHCPciPageCtrl *ctrl, const KPHCExtentLocation location, bool need_write)
{
    if (need_write) {
        int nbytes = FileWrite(location.fd, (char *)ctrl->pciPage, BLCKSZ, location.headerNum * BLCKSZ,
                                (uint32)WAIT_EVENT_DATA_FILE_WRITE);
        if (nbytes != BLCKSZ) {
            (void)pg_atomic_fetch_sub_u32(&ctrl->refNum, 1);
            ctrl->loadStatus = CTRL_PAGE_LOADED_ERROR;
            LWLockRelease(ctrl->contentLock);
            ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
                errmsg("Failed to KPHCPciBufFreePage %s", FilePathName(location.fd))));
            return;
        }
        ctrl->loadStatus = CTRL_PAGE_IS_LOADED;
    }
    (void)pg_atomic_fetch_sub_u32(&ctrl->refNum, 1);
    LWLockRelease(ctrl->contentLock);
}

Size KPHCPciBufferSize()
{
    return 1024*1024*1024;
}

uint32 KPHCPciLockCount()
{
    Size buffer_size = KPHCPciBufferSize();
    uint32 ctrlCount = (uint32)((buffer_size - sizeof(KPHCPciPageBuffCtx)) /
        (sizeof(KPHCPciPageCtrl) + BLCKSZ + 3 * sizeof(KPHCPciHashBucket)));

    return (PCI_LRU_LIST_NUM + ctrlCount + ctrlCount * 3);
}

void KPHCPciBufInitCtx()
{
    char *buf;
    bool found = false;
    Size buffer_size = KPHCPciBufferSize();
    char* pciBufferBlocks =
            (char *)CACHELINEALIGN(ShmemInitStruct("PCI Buffer Blocks", buffer_size, &found));
    if (found) {
        return;
    }
    buf = pciBufferBlocks;
    memset((void*)buf, 0, sizeof(KPHCPciPageBuffCtx));
    gPciBufCtx = (KPHCPciPageBuffCtx *)(void *)buf;
    gPciBufCtx->maxCount = (uint32)((buffer_size - sizeof(KPHCPciPageBuffCtx)) /
        (sizeof(KPHCPciPageCtrl) + BLCKSZ + 3 * sizeof(KPHCPciHashBucket)));
    gPciBufCtx->ctrlBuf = (KPHCPciPageCtrl *)(void *)(buf + sizeof(KPHCPciPageBuffCtx));
    gPciBufCtx->pageBuf =
        ((char *)(gPciBufCtx->ctrlBuf)) + (Size)((Size)gPciBufCtx->maxCount * sizeof(KPHCPciPageCtrl));
    gPciBufCtx->hashtbl.buckets = (KPHCPciHashBucket *)(void *)(
        ((char *)(gPciBufCtx->pageBuf )) + (Size)((Size)gPciBufCtx->maxCount * BLCKSZ));

    for (uint32 i = 0; i < PCI_PART_LIST_NUM; i++) {
        gPciBufCtx->mainLru[i].lock = LWLockAssign((int)LWTRANCHE_PCI_BUFFER_CONTENT);
        LWLockInitialize(gPciBufCtx->mainLru[i].lock, (int)LWTRANCHE_PCI_BUFFER_CONTENT);
        gPciBufCtx->freeLru[i].lock = LWLockAssign((int)LWTRANCHE_PCI_BUFFER_CONTENT);
        LWLockInitialize(gPciBufCtx->freeLru[i].lock, (int)LWTRANCHE_PCI_BUFFER_CONTENT);
    }
    for (uint32 i = 1; i <= gPciBufCtx->maxCount; i++) {
        KPHCPciPageCtrl *ctrl = PCI_GET_CTRL_BY_ID(gPciBufCtx, i);
        memset((void*)ctrl,  0, sizeof(KPHCPciPageCtrl));
        ctrl->ctrlId = i;
        ctrl->pciPage = (KPHCCfsExtentHeader *)(void *)(gPciBufCtx->pageBuf + (Size)((Size)(i - 1) * BLCKSZ));
        ctrl->contentLock = LWLockAssign((int)LWTRANCHE_PCI_BUFFER_CONTENT);
        LWLockInitialize(ctrl->contentLock, (int)LWTRANCHE_PCI_BUFFER_CONTENT);
        KPHCPciLruPushNoLock(&gPciBufCtx->freeLru[ctrl->ctrlId % PCI_PART_LIST_NUM], ctrl, CTRL_STATE_FREE);
    }
    gPciBufCtx->hashtbl.hash = tag_hash;
    gPciBufCtx->hashtbl.match = memcmp;
    gPciBufCtx->hashtbl.bucketNum = 3 * gPciBufCtx->maxCount;
    for (uint32 i = 1; i <= gPciBufCtx->hashtbl.bucketNum; i++) {
        KPHCPciHashBucket *bucket = PCI_GET_BUCKET_BY_ID(gPciBufCtx, i);
        memset((void*)bucket, 0, sizeof(KPHCPciHashBucket));
        bucket->bckId = i;
        bucket->lock = LWLockAssign((int)LWTRANCHE_PCI_BUFFER_CONTENT);
        LWLockInitialize(bucket->lock, (int)LWTRANCHE_PCI_BUFFER_CONTENT);
    }
}